package com.lagou.java;

import com.lagou.java.handler.UserServiceHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.tomcat.jni.Address;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NettyServer {

    private static ExecutorService executor =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    public static ZkClient client = null;

    public static final String basePath = "/netty_provider";

    public static Map<Integer, String> cacheNodes = new HashMap<>();

    public static Map<Integer, UserServiceHandler> handlers = new ConcurrentHashMap<>();

    public static void initZkClient() throws Exception {
        // 创建连接
        client = new ZkClient("127.0.0.1:2181");
        System.out.println("Zookeeper session established. address is :" + "127.0.0.1:2181");
        // 创建父节点
        if (client.exists(basePath)) {
            client.deleteRecursive(basePath);
        }
        client.createPersistent(basePath, true);
        System.out.println("==>> 父节点：" + basePath + ", 创建成功！");
        // 注册监听事件
        client.subscribeChildChanges(basePath, new IZkChildListener() {
            public void handleChildChange(String parentPath,
                                          List<String> currentChilds) throws Exception {
                currentChilds = client.getChildren(parentPath);
                System.out.println("cache nodes :" + cacheNodes);
                System.out.println("current nodes :" + currentChilds);
                for (String node : currentChilds) {
                    if (!cacheNodes.values().contains(node)) {
                        try {
                            System.out.println(node + " > 新创建, 节点信息为: " + client.readData(parentPath + "/" + node));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                for (String node : cacheNodes.values()) {
                    if (!currentChilds.contains(node)) {
                        try {
                            System.out.println(node + " > 被删除");
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                // 清空缓存数据
                cacheNodes.clear();
                // 刷新缓存数据
                for (String node : currentChilds) {
                    String address = client.readData(basePath + "/" + node);
                    int port = Integer.parseInt(address.split(":")[1]);
                    cacheNodes.put(port, node);
                }

                System.out.println("cache nodes :" + cacheNodes);
                // 打印缓存数据
                System.out.println(parentPath + " 's child changed currentChilds:" + currentChilds);
            }
        });
    }


    public static ChannelFuture startServer (String hostName, int port){
        System.out.println("--------------");
        Channel channel = null;
        NioEventLoopGroup bossLoopGroup = null;
        NioEventLoopGroup childLoopGroup = null;
        ChannelFuture sync = null;
        try {
            bossLoopGroup = new NioEventLoopGroup();
            childLoopGroup = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
            UserServiceHandler userServiceHandler = new UserServiceHandler();

            bootstrap.group(bossLoopGroup, childLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
//                            pipeline.addLast(new StringDecoder());
                            // Netty半包解码器解决TCP粘包/拆包问题
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2));
                            // 返回结果编码码器
                            pipeline.addLast(new StringEncoder());
                            // 入参解码器
                            pipeline.addLast(new RpcDecoder());
                            // 自定义 ChannelHandler
                            pipeline.addLast(userServiceHandler);
                        }
                    });
            // 将 handler 放入缓存
            handlers.put(port, userServiceHandler);
            // 绑定端口
            sync = bootstrap.bind(hostName,port);
            com.lagou.java.ServerBootstrap.serverServices.put(port, sync);
            sync.sync();
            channel = sync.channel();
            if (channel != null) {
                System.out.println("========>>>>> netty server started at " + hostName + ":" + port);
                try {
                    // 将当前服务节点注册到 zookeeper
                    String path = client.createEphemeralSequential(basePath + "/provider", hostName + ":" + port);
                    System.out.println("==> path : " + path + "创建 成功！");
                } catch (Exception e) {
                    e.printStackTrace();
                }

            } else {
                System.out.println("========>>>>> netty server started failed ");
            }
            // 阻塞 main 线程
            ChannelFuture channelFuture = channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 服务端优雅关闭
            bossLoopGroup.shutdownGracefully();
            childLoopGroup.shutdownGracefully();
        }
        return sync;
    }


}
